
package com.huaweicloud.haydn.delivercore.agent.service.metadata.impl;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLTimeoutException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.ExcelWriter;
import com.alibaba.fastjson.JSON;
import com.huaweicloud.haydn.delivercore.agent.config.DatabaseContext;
import com.huaweicloud.haydn.delivercore.agent.dto.excel.DbConsistencyExcelDto;
import com.huaweicloud.haydn.delivercore.agent.dto.haydn.DataParseDto;
import com.huaweicloud.haydn.delivercore.agent.dto.haydn.DbConsistencyDto;
import com.huaweicloud.haydn.delivercore.agent.entity.ZipFileEntity;
import com.huaweicloud.haydn.delivercore.agent.service.metadata.ConsistencyMonitoringService;
import com.huaweicloud.haydn.delivercore.agent.service.metadata.DatabaseUtil;
import com.huaweicloud.haydn.delivercore.agent.utils.HiveAuthUtil;
import com.huaweicloud.haydn.delivercore.agent.utils.SSLUtils;
import com.huaweicloud.haydn.delivercore.agent.utils.ZipUtil;

import lombok.extern.slf4j.Slf4j;

import static com.huaweicloud.haydn.delivercore.agent.config.Constants.DOT_FILETYPE_CSV;
import static com.huaweicloud.haydn.delivercore.agent.config.Constants.FILETYPE_PROPERTIES;
import static com.huaweicloud.haydn.delivercore.agent.config.Constants.SYS_URL;

/**
 * 一致性检测（表行数）
 *
 * @since 2023-03-21
 */
@Service
@Slf4j
public class TableRowsService implements ConsistencyMonitoringService {

    @Autowired
    private DatabaseContext databaseContext;

    @Autowired
    private ZipUtil zipUtil;

    @PostConstruct
    public void postConstruct() {
        SSLUtils.disableSslVerification();
    }

    @Override
    public void genTableRowsData() {
        genDataType("consistencytablerows");
    }

    @Override
    public void genTableFieldData() {

    }

    @Override
    public void genTableContentData() {

    }

    public void genDataType(String type) {
        try (InputStream inputStream =
            new FileInputStream(System.getProperty(SYS_URL) + File.separator + "application." + FILETYPE_PROPERTIES)) {
            Properties monitorConfig = new Properties();
            monitorConfig.load(inputStream);
            List<ZipFileEntity> zipFileEntities = Collections.synchronizedList(new ArrayList<>());

            for (int i = 0; i < 100; i++) {
                if (!verifyProperties(monitorConfig, (i + 1))) {
                    log.info("break at i = " + i + ", idx = " + (i + 1));
                    break;
                }
                ZipFileEntity zipFile = genDataInPool(i + 1, monitorConfig, type);
                zipFileEntities.add(zipFile);
                log.info("Consistency comparison of table records for Group {} is complete!", i);
            }
            zipUtil.execZip(zipFileEntities, "monitorConsistencyResult");
        } catch (IOException e) {
            log.error(e.getMessage());
        }
    }

    /**
     * 校验数据库类型、ip、端口、用户名、密码是否为空
     */
    private boolean verifyProperties(Properties dataSourceConfig, int i) {
        if (StringUtils.isBlank(dataSourceConfig.getProperty("monitor.target.consistency.dbtype." + i))
            || StringUtils.isBlank(dataSourceConfig.getProperty("monitor.source.consistency.dbtype." + i))) {
            log.error("dbtype can not be null!");
            return false;
        }
        if (StringUtils.isBlank(dataSourceConfig.getProperty("monitor.target.consistency.hosts." + i))
            || StringUtils.isBlank(dataSourceConfig.getProperty("monitor.source.consistency.hosts." + i))) {
            log.error("hosts can not be null!");
            return false;
        }
        if (StringUtils.isBlank(dataSourceConfig.getProperty("monitor.target.consistency.port." + i))
            || StringUtils.isBlank(dataSourceConfig.getProperty("monitor.source.consistency.port." + i))) {
            log.error("port can not be null!");
            return false;
        }
        if (StringUtils.isBlank(dataSourceConfig.getProperty("monitor.target.consistency.username." + i))
            || StringUtils.isBlank(dataSourceConfig.getProperty("monitor.source.consistency.username." + i))) {
            log.error("username can not be null!");
            return false;
        }
        if (StringUtils.isBlank(dataSourceConfig.getProperty("monitor.target.consistency.password." + i))
            || StringUtils.isBlank(dataSourceConfig.getProperty("monitor.source.consistency.password." + i))) {
            log.error("password can not be null!");
            return false;
        }
        return true;
    }

    /**
     * 原则：与一个数据库同时只能有一个连接
     */
    private ZipFileEntity genDataInPool(int idx, Properties dataSourceConfig, String type) {
        List<DbConsistencyDto> dbConsistencyDtos =
            JSON.parseArray(dataSourceConfig.getProperty("monitor.consistency.mapping." + idx), DbConsistencyDto.class);
        ZipFileEntity zipFile = new ZipFileEntity();
        String fileName =
            idx + "_" + type + "_" + new SimpleDateFormat("yyyyMMdd").format(new Date()) + DOT_FILETYPE_CSV;
        zipFile.setFileName(fileName);
        zipFile.setType(type);
        if (CollectionUtils.isEmpty(dbConsistencyDtos)) {
            return zipFile;
        }

        List<String> sDbNameList =
            dbConsistencyDtos.stream().map(DbConsistencyDto::getSDb).distinct().collect(Collectors.toList());
        List<String> tDbNameList =
            dbConsistencyDtos.stream().map(DbConsistencyDto::getTDb).distinct().collect(Collectors.toList());

        String sUserName = dataSourceConfig.getProperty("monitor.source.consistency.username." + idx);
        String sPassword = dataSourceConfig.getProperty("monitor.source.consistency.password." + idx);
        String sDbType = dataSourceConfig.getProperty("monitor.source.consistency.dbtype." + idx);
        String sHosts = dataSourceConfig.getProperty("monitor.source.consistency.hosts." + idx);
        String sPort = dataSourceConfig.getProperty("monitor.source.consistency.port." + idx);
        String sIsSafeMode = dataSourceConfig.getProperty("monitor.source.consistency.safemode." + idx);

        String tUserName = dataSourceConfig.getProperty("monitor.target.consistency.username." + idx);
        String tPassword = dataSourceConfig.getProperty("monitor.target.consistency.password." + idx);
        String tDbType = dataSourceConfig.getProperty("monitor.target.consistency.dbtype." + idx);
        String tHosts = dataSourceConfig.getProperty("monitor.target.consistency.hosts." + idx);
        String tPort = dataSourceConfig.getProperty("monitor.target.consistency.port." + idx);
        String tIsSafeMode = dataSourceConfig.getProperty("monitor.target.consistency.safemode." + idx);
        log.info("Successfully read the configuration file and successfully initialized!");

        Connection sConnection = null;
        Connection tConnection = null;
        List<DbConsistencyExcelDto> excelDtos = new ArrayList<>();
        try {
            for (String sDbName : sDbNameList) { // 源端同时只能存在一个Connection
                sConnection = initDbUtil(sDbType, sIsSafeMode, sDbName, sHosts, sPort, sUserName, sPassword);
                List<DbConsistencyDto> consistencyDtos =
                    dbConsistencyDtos.stream().filter(dto -> sDbName.equals(dto.getSDb())).collect(Collectors.toList());
                for (String tDbName : tDbNameList) { // 目的端同时只能存在一个Connection
                    tConnection = initDbUtil(tDbType, tIsSafeMode, tDbName, tHosts, tPort, tUserName, tPassword);
                    setHudiVision(tConnection, tDbType);
                    List<DbConsistencyDto> consistencyDtoList = consistencyDtos.stream()
                        .filter(dto -> tDbName.equals(dto.getTDb()))
                        .collect(Collectors.toList());
                    if (consistencyDtoList.size() == 0) {
                        continue;
                    }
                    genDbConsistencyExcelDto(consistencyDtoList, excelDtos, type, sDbType, tDbType, sConnection,
                        tConnection);
                    closeConnection(tConnection);
                }
                closeConnection(sConnection);
            }
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            if (tConnection != null) {
                try {
                    tConnection.close();
                } catch (SQLException e) {
                    log.error("close connection failed!", e);
                }
            }
            if (sConnection != null) {
                try {
                    sConnection.close();
                } catch (SQLException e) {
                    log.error("close connection failed!", e);
                }
            }
        }

        // 生成excel
        ExcelWriter excelWriter = EasyExcel.write(fileName, DbConsistencyExcelDto.class).build();
        excelWriter.write(excelDtos, EasyExcel.writerSheet("sheet").build());
        excelWriter.finish();
        return zipFile;
    }

    private Connection initDbUtil(String dbType, String isSafeMode, String dbName, String host, String port,
        String userName, String password) {
        DatabaseUtil databaseUtil = databaseContext.getStrategy(dbType);
        if ("MRS-Hudi".equals(dbType) && "true".equals(isSafeMode)) {
            dbName = HiveAuthUtil.getHiveAuthJDBC(userName, dbName);
        }
        databaseUtil.init(host, port, dbName, userName, password, dbType, "", "");
        return databaseUtil.getConnection();
    }

    private void closeConnection(Connection connection) {
        if (connection == null) {
            return;
        }
        try {
            connection.close();
        } catch (SQLException e) {
            log.error("close connection failed!", e);
        }
    }

    private void genDbConsistencyExcelDto(List<DbConsistencyDto> dbConsistencyDtos,
        List<DbConsistencyExcelDto> excelDtos, String type, String sDbType, String tDbType, Connection sConnection,
        Connection tConnection) {
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        for (DbConsistencyDto consistencyDto : dbConsistencyDtos) {
            DbConsistencyExcelDto excelDto = new DbConsistencyExcelDto();
            excelDto.setMonitoringObjectUuid(consistencyDto.getUuid());
            String timestamp = consistencyDto.getTimestamp();
            String sourceDbName = consistencyDto.getSDb();
            String sourceSchemaName = consistencyDto.getSSchema();
            String sourceTableName = consistencyDto.getSTable();
            String sumField = consistencyDto.getSumField();
            String sourceWhere = consistencyDto.getSWhere();
            excelDto.setSourceDbName(sourceDbName);
            excelDto.setSourceSchemaName(sourceSchemaName);
            excelDto.setSourceTableName(sourceTableName);
            String sourceSql = getDbTableRowsSqlByDbType(sDbType, sourceDbName, sourceSchemaName, sourceTableName,
                timestamp, sumField, sourceWhere);
            String sourceRows = getDbRows(sourceSql, sConnection, sumField);
            if (StringUtils.isNumeric(sourceRows)) {
                excelDto.setSourceTableRow(sourceRows);
            } else {
                excelDto.setSourceException(sourceRows);
            }
            excelDto.setFromSql(sourceSql);
            log.info("source table name: {}", sourceTableName);

            String targetDbName = consistencyDto.getTDb();
            String targetSchemaName = consistencyDto.getTSchema();
            String targetTableName = consistencyDto.getTTable();
            excelDto.setTargetDbName(targetDbName);
            excelDto.setTargetSchemaName(targetSchemaName);
            excelDto.setTargetTableName(targetTableName);
            String targetWhere = consistencyDto.getTWhere();
            String targetSql = getDbTableRowsSqlByDbType(tDbType, targetDbName, targetSchemaName, targetTableName,
                timestamp, sumField, targetWhere);
            String targetRows = getDbRows(targetSql, tConnection, sumField);
            if (StringUtils.isNumeric(targetRows)) {
                excelDto.setTargetTableRow(targetRows);
            } else {
                excelDto.setTargetException(targetRows);
            }
            excelDto.setToSql(targetSql);
            excelDto.setMonitoringDate(format.format(new Date()));
            log.info("target table name: {}", targetTableName);
            excelDtos.add(excelDto);
        }
    }

    private String getDbTableRowsSqlByDbType(String dbType, String dbNm, String schemaNm, String tableNm,
        String timestampNm, String sumField, String where) { // fixme：带时间戳的
        if (StringUtils.isBlank(where)) {
            where = "1=1";
        }
        String sql = "";
        if ("mysql".equals(dbType)) {
            sql = String.format("select count(*) from %s.%s", dbNm, tableNm);
            if (StringUtils.isNotEmpty(timestampNm)) {
                sql = sql + " where " + timestampNm + " <= ?";
            }
        } else if ("MRS-Hudi".equals(dbType)) {
            if (StringUtils.isNotBlank(sumField)) {
                sql = String.format("select sum(%s) from %s where %s", sumField, tableNm, where);
            } else {
                sql = String.format("select count(1) from %s where %s", tableNm, where);
            }
        } else if ("dws".equals(dbType)) {
            sql = String.format("select count(*) from %s.%s", schemaNm, tableNm);
            if (StringUtils.isNotEmpty(timestampNm)) {
                sql = sql + " where " + "\"" + timestampNm + "\"" + " <= ?";
            }
        } else if ("oracle".equals(dbType)) {
            if (StringUtils.isNotBlank(sumField)) {
                sql = String.format("select sum(%s) from %s.%s where %s", sumField, schemaNm, "\"" + tableNm + "\"",
                    where);
            } else {
                sql = String.format("select count(*) from %s.%s where %s", schemaNm, "\"" + tableNm + "\"", where);
            }
        }
        return sql;
    }

    private String getDbRows(String sql, Connection conn, String sumField) {
        ResultSet rs = null;
        String tableRow = null;
        PreparedStatement ps = null;
        try {
            ps = conn.prepareStatement(sql, ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY);
            ps.setQueryTimeout(100);
            rs = ps.executeQuery();
            while (rs.next()) {
                if (StringUtils.isNotBlank(sumField)) {
                    tableRow = String.valueOf(rs.getLong(1));
                } else {
                    tableRow = rs.getString(1);
                }
            }
        } catch (SQLTimeoutException e) {
            return "SQL Timeout Exception";
        } catch (SQLException e) {
            log.error(e.getMessage(), e);
            return e.getMessage();
        } catch (Exception e) {
            log.info("Exception: ", e);
            return e.getMessage();
        } finally {
            if (rs != null) {
                try {
                    rs.close();
                } catch (SQLException e) {
                    log.error("close ResultSet failed: ", e);
                }
            }
            if (ps != null) {
                try {
                    ps.close();
                } catch (SQLException e) {
                    log.error("close PreparedStatement failed: ", e);
                }
            }
        }
        return tableRow;
    }

    private void closeResultSet(ResultSet resultSet) {
        if (resultSet == null) {
            return;
        }
        try {
            resultSet.close();
        } catch (SQLException e) {
            log.error("close ResultSet failed: ", e);
        }
    }

    private void closePreparedStatement(PreparedStatement preparedStatement) {
        if (preparedStatement == null) {
            return;
        }
        try {
            preparedStatement.close();
        } catch (SQLException e) {
            log.error("close PreparedStatement failed: ", e);
        }
    }

    private String getUtf8(String s) {
        if (org.apache.commons.lang3.StringUtils.isBlank(s)) {
            return s;
        }
        return new String(s.getBytes(StandardCharsets.ISO_8859_1), StandardCharsets.UTF_8);
    }

    private void setHudiVision(Connection tConnection, String dbType) {
        if (!"MRS-Hudi".equals(dbType)) {
            return;
        }
        ResultSet rs = null;
        PreparedStatement ps = null;
        try {
            ps = tConnection.prepareStatement("SET hive.execution.engine=mr", ResultSet.TYPE_SCROLL_INSENSITIVE,
                ResultSet.CONCUR_READ_ONLY);
            rs = ps.executeQuery();
            log.info("SET hive.execution.engine=mr");
        } catch (SQLException e) {
            log.error(e.getMessage(), e);
        } catch (Exception e) {
            log.info("Exception: ", e);
        } finally {
            if (rs != null) {
                try {
                    rs.close();
                } catch (SQLException e) {
                    log.error("close ResultSet failed: ", e);
                }
            }
            if (ps != null) {
                try {
                    ps.close();
                } catch (SQLException e) {
                    log.error("close PreparedStatement failed: ", e);
                }
            }
        }
    }

    @Override
    public void consistencyParseAndUpload(DataParseDto parseDto) {

    }

    @Override
    public void reportMonitorConsistencyData() {

    }
}
